/*
* Copyright (C) 2020 ~ 2023 chaigec All Rights Reserved.
*
* Author:     chaigec <chaigec@163.com>
* Maintainer: chaigec <chaigec@163.com>

 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
*/

#include "MkMediaServer.h"
#include "MkMediaServerConfig.h"
#include <MkMedia/MkFileSource.h>
#include <MkMedia/MkMediaUtil.h>
#include <MkPs/MkPsFileSource.h>
#include <MkFlv/MkFlvFileSource.h>
#include <MkRtmp/MkRtmpPlayer.h>
#include <MkRtsp/MkRtspServer.h>
#include <MkUtil/MkMd5.h>
#include <MkGbMedia/MkGbMediaServer.h>
#include <MkHttpMedia/MkWsFlvPlayer.h>
#include <MkHttp/MkHttpSession.h>
#include <MkRtp/MkRtpUdpServer.h>
#include <MkRtmp/MkRtmpServer.h>
#include <MkRtp/MkRtpServerPool.h>
#include <MkRtsp/MkRtspPlayer.h>
#include <MkUtil/MkLog.h>
#include <MkEvent/MkTimer.h>
#include <MkEvent/MkEventPool.h>
#include <MkEvent/MkTimerPool.h>
#include <MkEvent/MkEvent.h>
#include <MkUtil/MkFile.h>
#include <MkUtil/MkUtil.h>
#include <signal.h>

//config appkeys
#define ExeName                         "MkMediaServer"
#define HttpPort                        "http.port"
#define HttpUserName                    "http.userName"
#define HttpPassword                    "http.password"
#define HttpApiUriPrefix                "http.apiUriPrefix"
#define HttpKeepAliveSec                "http.keepAliveSec"
#define HttpMaxKeepAliveReq             "http.maxKeepAliveReq"
#define GbRealm                         "gb.realm"
#define GbHost                          "gb.host"
#define GbPort                          "gb.port"
#define GbMappingHost                   "gb.mappingHost"
#define GbMappingPort                   "gb.mappingPort"
#define GbRegServerId                   "gb.regServerId"
#define GbRegServerRealm                "gb.regServerRealm"
#define GbRegHost                       "gb.regHost"
#define GbRegPort                       "gb.regPort"
#define GbTcpSip                        "gb.tcpSip"
#define GbRegUserName                   "gb.regUserName"
#define GbRegPassword                   "gb.regPassword"
#define GbRegExpires                    "gb.regExpires"
#define GbHBInterval                    "gb.hbInterval"
#define GbHBCount                       "gb.hbCount"
#define RtpBeginPort                    "rtp.BeginPort"
#define RtpRtcpCheck                    "rtp.rtcpCheck"
#define RtpRtcpCheckTimeOut             "rtp.rtcpCheckTimeOut"
#define GeneralServerId                 "general.serverId"
#define GeneralNoneReaderCloseDelayMs   "general.noneReaderCloseDelayMs"  //mast more than 10*1000  maxint32 not auto close
#define GeneralStreamSecondOn           "general.streamSecondOn"
#define GeneralNotifyMediaInfo          "general.notifyMediaInfo"
#define GeneralNotifyMediaInfoInterval  "general.notifyMediaInfoInterval"  //milliseconds mast more than 60*1000
#define GeneralAppName                  "general.appName"
#define RtmpPort                        "rtmp.port"
#define RtspPort                        "rtsp.port"
#define TcpUnSendMs                     "tcp.unSendMilliSec"
#define TcpUnSendMB                     "tcp.unSendMB"
#define UdpUnSendMs                     "udp.unSendMilliSec"
#define UdpUnSendMB                     "udp.unSendMB"
#define WsMaxConnectorSize              "ws.maxConnectorSize"
#define WsAccessKey                     "ws.accessKey"
#define WsUrl                           "ws.url"

CMkMediaServer *g_pMediaServer = nullptr;
CMkMediaServer::CMkMediaServer(CMkEventPool* pEventPool, CMkTimer* pTimer)
    : m_pEventPool(pEventPool)
    , m_pTimer(pTimer)
    , m_pRtmpServer(nullptr)
    , m_pHttpMediaServer(nullptr)
    , m_pGbMediaServer(nullptr)
    , m_pRtpPool(nullptr)
    , m_pRtspServer(nullptr)
    , m_bReboot(FALSE)
{
    g_pMediaServer = this;
    AddParser("version", MkHttpMethodGet, &CMkMediaServer::GetVersionParse);

    AddParser("config", MkHttpMethodGet, &CMkMediaServer::GetConfigParse);
    AddParser("config", MkHttpMethodPut, &CMkMediaServer::SetConfigParse);
    AddParser("ws-access-key", MkHttpMethodGet, &CMkMediaServer::GetWsAccessKeyParse);

    AddParser("stream-proxy", MkHttpMethodPost, &CMkMediaServer::AddProxyStreamParse);
    AddParser("media-source", MkHttpMethodGet, &CMkMediaServer::GetMediaSourceParse);
    AddParser("media-source", MkHttpMethodDelete, &CMkMediaServer::DeleteMediaSourceParse);
    AddParser("media-consumer", MkHttpMethodGet, &CMkMediaServer::GetMediaConumerParse);
    AddParser("rtp-port-range", MkHttpMethodGet, &CMkMediaServer::GetRtpPortRangeParse);
    AddParser("shutdown", MkHttpMethodPut, &CMkMediaServer::ShutDownParse);
    AddParser("reboot", MkHttpMethodPut, &CMkMediaServer::RebootParse);

    m_MediaCbs.SourceAddCb = [this](CMkMediaSource* pSource) {
        return OnMediaAdd(pSource);
    };
    m_MediaCbs.SourceDeleteCb = [this](const MkString& SourceSession) {
        return OnMediaDelete(SourceSession);
    };
    m_MediaCbs.ConsumerAddCb = [this](const MkString& SourceSession, CMkMediaConsume* pConsume, CMkMediaConsumer* pConsumer) {
        return OnMediaConsumerAdd(SourceSession, pConsume, pConsumer);
    };
    m_MediaCbs.ConsumerClosedCb = [this](const MkString& SourceSession, const MkMediaConsumeType& ConsumeType, const MkString& ConsumerSession) {
        return OnMediaConsumerClosed(SourceSession, ConsumeType, ConsumerSession);
    };

    m_MediaCbs.ConsumerPlayCtrlCb = [this](CMkMediaConsumer* pConsumer, const MkPlaybackControlType& ControlType, float fSpeed, Uint32 offsetSec) {
        return OnMediaConsumerPlayCtrl(pConsumer, ControlType, fSpeed, offsetSec);
    };
    m_MediaCbs.GetSourceParamCb = [this](const MkString& Path, CMkVideoParam& VideoParam, CMkAudioParam& AudioParam, Uint32& Duration, Uint32& FileSize, MkString& SourceSession) {
        return OnMediaSourceGetParam(Path, VideoParam, AudioParam, Duration, FileSize, SourceSession);
    };
}


Uint32 CMkMediaServer::Init(int argc, char** argv)
{
#ifndef _WIN32
    signal(SIGPIPE, SIG_IGN);  //ignore sigpipe
#endif
    signal(SIGINT, [](int sig) {
        if (sig == SIGINT && g_pMediaServer->GetTimer()) {
            g_pMediaServer->GetTimer()->LoopBreak();
        }
    });// set ctrl +c
    MkLogChangeLevel(LOG_DEBUG);
    MkLogInitLevel(ExeName, TO_SCREEN | TO_FILE);
    Uint32 ErrorCode = NoneError;
    do {
        m_bReboot = FALSE;
        ErrorCode = m_pEventPool->Start();
        MkCheckErrorLog(ErrorCode, "start event pool failed\n");
        ErrorCode = Start();
        MkCheckErrorLog(ErrorCode, "start media server failed\n");
        GetTimer()->Dispatch();
        CMkThread::ThreadSleep(1000); //wait all stopped
        m_pEventPool->Stop();
        Stop();
    } while (m_bReboot);
    return NoneError;
}

Uint32 CMkMediaServer::Start()
{
    Uint32 ErrorCode = NoneError;
    ErrorCode = ReadConfig();
    MkCheckErrorLog(ErrorCode, "read config failed\n");

    CMkSocket::SetTcpSendFailCumulativeMs(m_ConfigIni.GetProfileUint32(TcpUnSendMs, 3000));
    CMkSocket::SetTcpSendFailCumulativeMB(m_ConfigIni.GetProfileUint32(TcpUnSendMB, 20));
    CMkSocket::SetUdpSendFailCumulativeMB(m_ConfigIni.GetProfileUint32(UdpUnSendMB, 20));
    CMkSocket::SetUdpSendFailCumulativeMs(m_ConfigIni.GetProfileUint32(UdpUnSendMs, 3000));
    CMkMediaSource::SetCachePacket(m_ConfigIni.GetProfileBoolean(GeneralStreamSecondOn));

    ErrorCode = StartRtpPool();
    MkCheckErrorLog(ErrorCode, "start rtp pool failed\n");

    ErrorCode = StartHttpServer();
    MkCheckErrorLog(ErrorCode, "start http server failed\n");

    ErrorCode = StartGbServer();
    MkCheckErrorLog(ErrorCode, "start sip server failed\n");

    ErrorCode = StartRtmpServer();
    MkCheckErrorLog(ErrorCode, "start rtmp server failed\n");

    ErrorCode = StartRtspServer();
    MkCheckErrorLog(ErrorCode, "start rtsp server failed\n");

    if (m_ConfigIni.GetProfileBoolean(GeneralNotifyMediaInfo)) {
        Uint32 Interval = MkMax(m_ConfigIni.GetProfileUint32(GeneralNotifyMediaInfoInterval), 60 * 1000);
        GetTimer()->AddTimer([this]() {
            WsNotifyMediaInfo();
        }, Interval, m_NotifyMediaInfoTimer);
    }

    MkString strVersion;
    GetVersion(strVersion);
    MkInfoLog("start mk media server version:%s\n", strVersion.c_str());
    return NoneError;
}

Uint32 CMkMediaServer::Stop()
{
    if (m_pHttpMediaServer) {
        m_pHttpMediaServer->StopServer();
    }
    if (m_pGbMediaServer) {
        m_pGbMediaServer->Stop();
    }
    if (m_pRtspServer) {
        m_pRtspServer->StopServer();
    }
    if (m_pRtmpServer) {
        m_pRtmpServer->StopServer();
    }
    return NoneError;
}

Uint32 CMkMediaServer::ReadConfig()
{
    MkString ConfigPath;
    CMkFile::GetConfigPath(ConfigPath);
    if (!CMkFile::IsDir(ConfigPath)) {
        CMkFile::MakeDir(ConfigPath);
    }
    MkString ConfigFile = ConfigPath + "/" + ExeName + ".ini";
#ifdef MkUt
    CMkFile::DeleteLocalFile(ConfigFile);
#endif
    Uint32 ErrorCode = m_ConfigIni.LoadFile(ConfigFile);
    if (NoneError != ErrorCode) {
#ifdef MkUt
        Uint32 RandomPort;
        while (1) {
            RandomPort = CMkUtil::GetRandomNumber() & 0xFFFF;
            if (NoneError == CMkUtil::CheckTcpIsUseAbled(RandomPort)) {
                m_ConfigIni.SetProfileUint32(HttpPort, RandomPort);
                break;
            }
        }

        while (1) {
            RandomPort = CMkUtil::GetRandomNumber() & 0xFFFF;
            if (NoneError == CMkUtil::CheckTcpIsUseAbled(RandomPort)) {
                m_ConfigIni.SetProfileUint32(RtspPort, RandomPort);
                break;
            }
        }

        while (1) {
            RandomPort = CMkUtil::GetRandomNumber() & 0xFFFF;
            if (NoneError == CMkUtil::CheckTcpIsUseAbled(RandomPort)) {
                m_ConfigIni.SetProfileUint32(RtmpPort, RandomPort);
                break;
            }
        }

        m_ConfigIni.SetProfileString(HttpUserName, "admin");
        m_ConfigIni.SetProfileString(HttpPassword, "admin");
        m_ConfigIni.SetProfileString(WsUrl, "/ws/notify");
        m_ConfigIni.SetProfileString(HttpApiUriPrefix, "/api/v1");
        m_ConfigIni.SetProfileUint32(HttpKeepAliveSec, 60);
        m_ConfigIni.SetProfileUint32(HttpMaxKeepAliveReq, 1000);
        m_ConfigIni.SetProfileString(GeneralServerId, "34020000002020000001");
        m_ConfigIni.SetProfileString(GbRealm, "3402000000");
        m_ConfigIni.SetProfileString(GbHost, "");
        while (1) {
            RandomPort = CMkUtil::GetRandomNumber() & 0xFFFF;
            if (NoneError == CMkUtil::CheckTcpIsUseAbled(RandomPort)
                && NoneError == CMkUtil::CheckUdpIsUseAbled(RandomPort)) {
                m_ConfigIni.SetProfileUint32(GbPort, RandomPort);
                break;
            }
        }
        m_ConfigIni.SetProfileString(GbMappingHost, "");
        m_ConfigIni.SetProfileUint32(GbMappingPort, 0);

        m_ConfigIni.SetProfileUint32(RtpBeginPort, 60000);
        m_ConfigIni.SetProfileUint32(TcpUnSendMs, 3000);
        m_ConfigIni.SetProfileUint32(TcpUnSendMB, 20);
        m_ConfigIni.SetProfileUint32(UdpUnSendMs, 3000);
        m_ConfigIni.SetProfileUint32(UdpUnSendMB, 20);
#else
        m_ConfigIni.SetProfileUint32(HttpPort, 80);
        m_ConfigIni.SetProfileString(HttpUserName, "admin");
        m_ConfigIni.SetProfileString(HttpPassword, "admin");
        m_ConfigIni.SetProfileString(HttpApiUriPrefix, "/api/v1");
        m_ConfigIni.SetProfileUint32(HttpKeepAliveSec, 60);
        m_ConfigIni.SetProfileUint32(HttpMaxKeepAliveReq, 1000);
        m_ConfigIni.SetProfileString(GeneralServerId, "34020000002020000001");
        m_ConfigIni.SetProfileUint32(GeneralNoneReaderCloseDelayMs, 30000);
        m_ConfigIni.SetProfileBoolean(GeneralStreamSecondOn, TRUE);
        m_ConfigIni.SetProfileBoolean(GeneralNotifyMediaInfo, TRUE);
        m_ConfigIni.SetProfileUint32(GeneralNotifyMediaInfoInterval, 60000);
        m_ConfigIni.SetProfileString(GeneralAppName, "live");
        m_ConfigIni.SetProfileString(GbRealm, "3402000000");
        m_ConfigIni.SetProfileString(GbHost, "");
        m_ConfigIni.SetProfileUint32(GbPort, 5080);
        m_ConfigIni.SetProfileString(GbMappingHost, "");
        m_ConfigIni.SetProfileUint32(GbMappingPort, 0);
        m_ConfigIni.SetProfileString(GbRegServerId, "34020000002000000001");
        m_ConfigIni.SetProfileString(GbRegServerRealm, "3402020000");
        m_ConfigIni.SetProfileString(GbRegHost, "192.168.100.120");
        m_ConfigIni.SetProfileUint32(GbRegPort, 5060);
        m_ConfigIni.SetProfileBoolean(GbTcpSip, FALSE);
        m_ConfigIni.SetProfileString(GbRegUserName, "34020000002020000001");
        m_ConfigIni.SetProfileString(GbRegPassword, "12345678");
        m_ConfigIni.SetProfileUint32(GbRegExpires, 0);
        m_ConfigIni.SetProfileUint32(GbHBInterval, 6);
        m_ConfigIni.SetProfileUint32(GbHBCount, 3);
        m_ConfigIni.SetProfileUint32(RtmpPort, 1935);
        m_ConfigIni.SetProfileUint32(RtspPort, 554);
        m_ConfigIni.SetProfileUint32(RtpBeginPort, 60000);
        m_ConfigIni.SetProfileBoolean(RtpRtcpCheck, FALSE);
        m_ConfigIni.SetProfileUint32(RtpRtcpCheckTimeOut, 10);
        m_ConfigIni.SetProfileUint32(TcpUnSendMs, 3000);
        m_ConfigIni.SetProfileUint32(TcpUnSendMB, 20);
        m_ConfigIni.SetProfileUint32(UdpUnSendMs, 3000);
        m_ConfigIni.SetProfileUint32(UdpUnSendMB, 20);
        m_ConfigIni.SetProfileString(WsAccessKey, "MkMediaServer");
        m_ConfigIni.SetProfileUint32(WsMaxConnectorSize, 3);
        m_ConfigIni.SetProfileString(WsUrl, "/ws/notify");
#endif
        ErrorCode = m_ConfigIni.WriteFile(ConfigFile);
    }
    return ErrorCode;
}

Uint32 CMkMediaServer::GetVersion(MkString& Version)
{
    Version = MkString(MkMediaServerVersion) + "_build(" + MkString(__DATE__) + " " + MkString(__TIME__) + ")";
    return NoneError;
}

Uint32 CMkMediaServer::StartHttpServer()
{
    MkDelete(m_pHttpMediaServer);
    m_pHttpMediaServer = new CMkHttpMediaServer(m_pEventPool, m_pTimer, [this](const MkHttpMethod& Method, const MkString& Path, const MkMapString& MapQuery, const Json::Value& JsonReq, const MkHttpResponseJsonCb& cb) {
        return OnRequestJsonParse(Method, Path, MapQuery, JsonReq, cb);
    }, [this](CMkHttpSession* pSession) {
        OnWsAccessed(pSession);
    }, m_MediaCbs);
    m_pHttpMediaServer->SetUserPass(m_ConfigIni.GetProfileString(HttpUserName), m_ConfigIni.GetProfileString(HttpPassword));
    m_pHttpMediaServer->SetApiUriPrefix(m_ConfigIni.GetProfileString(HttpApiUriPrefix, "/api/v1"));
    m_pHttpMediaServer->SetWebSocketPath(m_ConfigIni.GetProfileString(WsUrl, "/ws/notify"));
    m_pHttpMediaServer->SetWebSocketAccessKey(m_ConfigIni.GetProfileString(WsAccessKey));
    m_pHttpMediaServer->SetMaxWsConnectorSize(m_ConfigIni.GetProfileUint32(WsMaxConnectorSize));
    m_pHttpMediaServer->SetKeepAliveSeconds(m_ConfigIni.GetProfileUint32(HttpKeepAliveSec, 60));
    m_pHttpMediaServer->SetMaxKeepAliveRequest(m_ConfigIni.GetProfileUint32(HttpMaxKeepAliveReq, 1000));
    Uint32  ErrorCode = m_pHttpMediaServer->StartServer(m_ConfigIni.GetProfileUint32(HttpPort, 80));
    MkCheckErrorLog(ErrorCode, "start http server port:%d failed", m_ConfigIni.GetProfileUint32(HttpPort, 80));
    return ErrorCode;
}

Uint32 CMkMediaServer::StartGbServer()
{
    MkDelete(m_pGbMediaServer);
    m_pGbMediaServer = new CMkGbMediaServer(m_pEventPool, m_pTimer, m_pRtpPool, [this]() {
        WsGbRegServerInfo();
    }, m_MediaCbs);
    MkGbStartParam Param;
    Param.LocalHost = m_ConfigIni.GetProfileString(GbHost);
    Param.LocalPort = m_ConfigIni.GetProfileUint32(GbPort);
    Param.MappingHost = m_ConfigIni.GetProfileString(GbMappingHost);
    Param.MappingPort = m_ConfigIni.GetProfileUint32(GbMappingPort);
    Param.RegisterExpires = 0;
    Param.SubscribeExpires = 0;
    Param.SipId = m_ConfigIni.GetProfileString(GeneralServerId);
    Param.SipRealm = m_ConfigIni.GetProfileString(GbRealm);
    Param.bCheckDstId = FALSE; //media server not check sip id
    Uint32 ErrorCode = m_pGbMediaServer->Start(Param, m_ConfigIni.GetProfileString(GeneralAppName));
    MkCheckErrorLog(ErrorCode, "start gb media server failed\n");
    if (0 != m_ConfigIni.GetProfileUint32(GbRegExpires)) {
        ErrorCode = m_pGbMediaServer->SetRegisterServer(m_ConfigIni.GetProfileString(GbRegServerId), m_ConfigIni.GetProfileString(GbRegServerRealm), m_ConfigIni.GetProfileString(GbRegHost), m_ConfigIni.GetProfileUint32(GbRegPort), m_ConfigIni.GetProfileBoolean(GbTcpSip), m_ConfigIni.GetProfileString(GbRegUserName), m_ConfigIni.GetProfileString(GbRegPassword), m_ConfigIni.GetProfileUint32(GbRegExpires), m_ConfigIni.GetProfileUint32(GbHBInterval), m_ConfigIni.GetProfileUint32(GbHBCount));
    }
    return NoneError;
}

Uint32 CMkMediaServer::StartRtspServer()
{
    MkDelete(m_pRtspServer);
    m_pRtspServer = new CMkRtspServer(m_pEventPool, m_pTimer, m_pRtpPool->GetVecRtpUdpServer(), [this](MkString& UserName, MkString& Password) {
        return OnGetAuth(UserName, Password);
    }, m_MediaCbs);
    Uint32 ErrorCode = m_pRtspServer->StartServer(m_ConfigIni.GetProfileUint32(RtspPort, 554));
    MkCheckErrorLog(ErrorCode, "start rtsp server port:%d failed\n", m_ConfigIni.GetProfileUint32(RtspPort, 554));
    MkInfoLog("start rtsp server port:%d success\n", m_ConfigIni.GetProfileUint32(RtspPort, 554));
    return NoneError;
}

Uint32 CMkMediaServer::StartRtmpServer()
{
    MkDelete(m_pRtmpServer);
    m_pRtmpServer = new CMkRtmpServer(m_pEventPool, m_pTimer, m_MediaCbs);

    Uint32 ErrorCode = m_pRtmpServer->StartServer(m_ConfigIni.GetProfileUint32(RtmpPort, 1935));
    MkCheckErrorLog(ErrorCode, "start rtmp server:%d failed\n", m_ConfigIni.GetProfileUint32(RtmpPort, 1935));
    MkInfoLog("start rtmp port:%d success\n", m_ConfigIni.GetProfileUint32(RtmpPort, 1935));
    return NoneError;
}

Uint32 CMkMediaServer::StartRtpPool()
{
    MkDelete(m_pRtpPool);
    CMkRtpUdpServer::SetRtcpCheckFlag(m_ConfigIni.GetProfileBoolean(RtpRtcpCheck));
    CMkRtpUdpServer::SetRtcpTimeOutSec(m_ConfigIni.GetProfileUint32(RtpRtcpCheckTimeOut));
    m_pRtpPool = new CMkRtpServerPool(m_pEventPool, m_pTimer);
    m_pRtpPool->SetBeginPort(m_ConfigIni.GetProfileUint32(RtpBeginPort, 60000));
    return m_pRtpPool->Start();
}

Uint32 CMkMediaServer::OnRequestJsonParse(const MkHttpMethod& Method, const MkString& Path, const MkMapString& MapQuery, const Json::Value& JsonReq, const MkHttpResponseJsonCb& cb)
{
    Uint32 ErrorCode = NoneError;
    MapParser TmpMapParser;
    switch (Method) {
    case MkHttpMethodDelete:
        TmpMapParser = m_MapDeleteParser;
        break;
    case MkHttpMethodGet:
        TmpMapParser = m_MapGetParser;
        break;
    case MkHttpMethodPut:
        TmpMapParser = m_MapPutParser;
        break;
    case MkHttpMethodPost:
        TmpMapParser = m_MapPostParser;
        break;
    default:
        ErrorCode = UnSupported;
        break;
    }
    MapParser::iterator it = TmpMapParser.find(Path);
    if (it != TmpMapParser.end()) {
        ErrorCode = (this->*(it->second))(JsonReq, MapQuery, cb);
    } else {
        Json::Value Support;
        it = TmpMapParser.begin();
        for (int i = 0; it != TmpMapParser.end(); it++, i++) {
            Support[i] = m_pHttpMediaServer->GetApiUriPrefix() + "/" + it->first;
        }
        cb(InvalidUrl, Support);
        return NoneError;
    }
    return ErrorCode;
}

Uint32 CMkMediaServer::OnMediaAdd(CMkMediaSource* pSource)
{
    if ("live" != pSource->GetAppName()) {
        return InvalidParams;
    }
    m_MediaSourceLock.WLock();
    m_MapMediaSource[pSource->GetSession()] = pSource;
    m_MediaSourceLock.WUnlock();
    Uint32 NoneReaderCloseDelayMs = MkMax(m_ConfigIni.GetProfileUint32(GeneralNoneReaderCloseDelayMs), 10 * 1000);
    if (MaxInt32 != NoneReaderCloseDelayMs) {
        MkString SourceSession = pSource->GetSession();
        GetTimer()->AddDelayer([this, SourceSession]() {
            DeleteSource(SourceSession);
        }, NoneReaderCloseDelayMs, SourceSession);
    }
    Json::Value JsonBody;
    JsonBody["appName"] = pSource->GetAppName();
    JsonBody["streamPath"] = pSource->GetStreamPath();
    JsonBody["session"] = pSource->GetSession();
    if (MkCodecUnknown != pSource->GetVideoParam().GetCodec()) {
        Json::Value JsonVideo;
        JsonVideo["frameRate"] = pSource->GetVideoParam().GetFrameRate();
        JsonVideo["width"] = pSource->GetVideoParam().GetWidth();
        JsonVideo["bitRate"] = pSource->GetVideoParam().GetBitRate();
        JsonVideo["height"] = pSource->GetVideoParam().GetHeight();
        JsonVideo["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(pSource->GetVideoParam().GetCodec());
        JsonBody["video"] = JsonVideo;
    }
    if (MkCodecUnknown != pSource->GetAudioParam().GetCodec()) {
        Json::Value JsonAudio;
        JsonAudio["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(pSource->GetAudioParam().GetCodec());
        JsonAudio["sampleRate"] = CMkMediaUtil::AudioSampleRateIndexToRate(pSource->GetAudioParam().GetSampleRate());
        JsonAudio["sampleSize"] = CMkMediaUtil::AudioSampleSizeIndexToSize(pSource->GetAudioParam().GetSampleSize());
        JsonAudio["bStereo"] = TRUE == pSource->GetAudioParam().GetStereo();
        JsonAudio["channel"] = pSource->GetAudioParam().GetChannel();
        JsonAudio["bitRate"] = pSource->GetAudioParam().GetBitRate();
        JsonBody["audio"] = JsonAudio;
    }
    return WsNotifyMsg("MediaSourceAdded", JsonBody);
}

Uint32 CMkMediaServer::OnMediaDelete(const MkString& SourceSession)
{
    DeleteSource(SourceSession);
    return NoneError;
}

Uint32 CMkMediaServer::OnMediaConsumerAdd(const MkString& SourceSession, CMkMediaConsume* pConsume, CMkMediaConsumer* pConsumer)
{
    Uint32 ErrorCode = InvalidSession;
    m_MediaSourceLock.WLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.find(SourceSession);
    if (it != m_MapMediaSource.end() && it->second) {
        pConsume->SetParentSessionId(SourceSession);
        pConsumer->SetParentSessionId(SourceSession);
        it->second->AddConsumer(pConsume, pConsumer);
        ErrorCode = NoneError;
    }
    m_MediaSourceLock.WUnlock();
    GetTimer()->DeleteDelayer(SourceSession);
    return ErrorCode;
}

Uint32 CMkMediaServer::OnMediaConsumerClosed(const MkString& SourceSession, const MkMediaConsumeType& ConsumeType, const MkString& ConsumerSession)
{
    BOOL bEmpty = FALSE;
    Json::Value JsonBody;
    Json::Value JsonNoneReaderSource;
    CMkMediaConsumer *pConsumer = nullptr;
    CMkTime NowTime = NowMkTime;
    m_MediaSourceLock.WLock();
    MkMap<MkString, CMkMediaSource*>::iterator it;
    if (SourceSession.empty()) {
        it = m_MapMediaSource.begin();
        for (; it != m_MapMediaSource.end(); it++) {
            if (NoneError == it->second->GetConsumer(ConsumeType, ConsumerSession, &pConsumer) && pConsumer) {
                JsonBody["consumerId"] = pConsumer->GetConsumerId();
                MkString StrTime;
                pConsumer->GetStartTime().GetString(StrTime);
                JsonBody["startTime"] = StrTime;
                JsonBody["sendBytes"] = pConsumer->GetSendBytes();
                JsonBody["averageSpeedBytes"] = CMkMediaUtil::GetAverageSpeed(pConsumer->GetSendBytes(), pConsumer->GetStartTime(), NowTime);
                it->second->DeleteConsumer(ConsumeType, ConsumerSession, bEmpty);
                if (bEmpty) {
                    JsonNoneReaderSource["sessionId"] = it->second->GetSession();
                    JsonNoneReaderSource["appName"] = it->second->GetAppName();
                    JsonNoneReaderSource["streamPath"] = it->second->GetStreamPath();
                }
                break;
            }
        }
    } else {
        it = m_MapMediaSource.find(SourceSession);
        if (it != m_MapMediaSource.end()) {
            if (NoneError == it->second->GetConsumer(ConsumeType, ConsumerSession, &pConsumer) && pConsumer) {
                JsonBody["ConsumerId"] = pConsumer->GetConsumerId();
                MkString StrTime;
                pConsumer->GetStartTime().GetString(StrTime);
                JsonBody["StartTime"] = StrTime;
                JsonBody["SendBytes"] = pConsumer->GetSendBytes();
                JsonBody["AverageSpeedBytes"] = CMkMediaUtil::GetAverageSpeed(pConsumer->GetSendBytes(), pConsumer->GetStartTime(), NowTime);
                it->second->DeleteConsumer(ConsumeType, ConsumerSession, bEmpty);
                if (bEmpty) {
                    JsonNoneReaderSource["sessionId"] = it->second->GetSession();
                    JsonNoneReaderSource["appName"] = it->second->GetAppName();
                    JsonNoneReaderSource["streamPath"] = it->second->GetStreamPath();
                }
            }
        }
    }
    m_MediaSourceLock.WUnlock();
    if (bEmpty) {
        // set -1 is not delete media source
        Uint32 NoneReaderCloseDelayMs = MkMax(m_ConfigIni.GetProfileUint32(GeneralNoneReaderCloseDelayMs), 10 * 1000);
        MkDebugLog("session:%s has none reader after %lu millsec shutdown it\n", SourceSession.c_str(), NoneReaderCloseDelayMs);
        if (MaxInt32 != NoneReaderCloseDelayMs) {
            MkString DelaySession = SourceSession;
            GetTimer()->AddDelayer([this, SourceSession]() {
                DeleteSource(SourceSession);
            }, NoneReaderCloseDelayMs, DelaySession);
        }
        // notify none reader
        WsNotifyMsg("MediaSourceNoneReader", JsonNoneReaderSource);
    }

    return WsNotifyMsg("ConsumerClosed", JsonBody);
}

Uint32 CMkMediaServer::OnMediaConsumerPlayCtrl(CMkMediaConsumer* pConsumer, const MkPlaybackControlType& ControlType, float fSpeed, Uint32 offsetSec)
{
    m_MediaSourceLock.RLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.find(pConsumer->GetParentSessionId());
    if (it != m_MapMediaSource.end()) {
        it->second->PlayControl(ControlType, fSpeed, offsetSec);
    }
    m_MediaSourceLock.RUnlock();
    return NoneError;
}

Uint32 CMkMediaServer::OnMediaSourceGetParam(const MkString& Path, CMkVideoParam& VideoParam, CMkAudioParam& AudioParam, Uint32& Duration, Uint32& FileSize, MkString& SourceSession)
{
    Uint32 ErrorCode = InvalidParams;
    m_MediaSourceLock.RLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.begin();
    for (; it != m_MapMediaSource.end(); it++) {
        if (Path == "/" + it->second->GetAppName() + "/" + it->second->GetStreamPath()) {
            VideoParam = it->second->GetVideoParam();
            AudioParam = it->second->GetAudioParam();
            SourceSession = it->first;
            Duration = it->second->m_Duration;
            FileSize = it->second->m_FileSize;
            ErrorCode = NoneError;
            break;
        }
    }
    m_MediaSourceLock.RUnlock();
    return ErrorCode;
}

Uint32 CMkMediaServer::OnGetAuth(MkString& UserName, MkString& Password)
{
    return NoneError;
}

void CMkMediaServer::OnWsAccessed(CMkHttpSession* pSession)
{
    WsNotifyConfig(pSession);
    WsGbRegServerInfo(pSession);
    WsNotifyMediaInfo(pSession);
}

Uint32 CMkMediaServer::DeleteSource(const MkString& SourceSession)
{
    Json::Value JsonBody;
    BOOL bFind = FALSE;
    m_MediaSourceLock.WLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.find(SourceSession);
    if (it != m_MapMediaSource.end()) {
        JsonBody[0]["appName"] = it->second->GetAppName();
        JsonBody[0]["streamPath"] = it->second->GetStreamPath();
        JsonBody[0]["session"] = it->second->GetSession();
        JsonBody[0]["recvBytes"] = it->second->GetRecvBytes();
        MkString StartTime;
        it->second->GetStartTime().GetString(StartTime);
        JsonBody[0]["startTime"] = StartTime;
        JsonBody[0]["averageSpeedByte"] = CMkMediaUtil::GetAverageSpeed(it->second->GetRecvBytes(), it->second->GetStartTime());
        if (MkCodecUnknown != it->second->GetVideoParam().GetCodec()) {
            Json::Value JsonVideo;
            JsonVideo["frameRate"] = it->second->GetVideoParam().GetFrameRate();
            JsonVideo["width"] = it->second->GetVideoParam().GetWidth();
            JsonVideo["bitRate"] = it->second->GetVideoParam().GetBitRate();
            JsonVideo["height"] = it->second->GetVideoParam().GetHeight();
            JsonVideo["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetVideoParam().GetCodec());
            JsonBody[0]["video"] = JsonVideo;
        }
        if (MkCodecUnknown != it->second->GetAudioParam().GetCodec()) {
            Json::Value JsonAudio;
            JsonAudio["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetAudioParam().GetCodec());
            JsonAudio["sampleRate"] = CMkMediaUtil::AudioSampleRateIndexToRate(it->second->GetAudioParam().GetSampleRate());
            JsonAudio["sampleSize"] = CMkMediaUtil::AudioSampleSizeIndexToSize(it->second->GetAudioParam().GetSampleSize());
            JsonAudio["bStereo"] = TRUE == it->second->GetAudioParam().GetStereo();
            JsonAudio["channel"] = it->second->GetAudioParam().GetChannel();
            JsonAudio["bitRate"] = it->second->GetAudioParam().GetBitRate();
            JsonBody[0]["audio"] = JsonAudio;
        }
        it->second->StopMediaSource();
        MkDebugLog("session removed\n", SourceSession.c_str());
        bFind = TRUE;
        m_MapMediaSource.erase(it);
    }
    m_MediaSourceLock.WUnlock();
    if (bFind) {
        WsNotifyMsg("MediaSourceClosed", JsonBody);
    }
    return NoneError;
}

Uint32 CMkMediaServer::ClearSource()
{
    Json::Value JsonBody;
    int Index = 0;
    m_MediaSourceLock.WLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.begin();
    for (; it != m_MapMediaSource.end(); it++, Index++) {
        JsonBody[Index]["appName"] = it->second->GetAppName();
        JsonBody[Index]["streamPath"] = it->second->GetStreamPath();
        JsonBody[Index]["session"] = it->second->GetSession();
        JsonBody[Index]["recvBytes"] = it->second->GetRecvBytes();
        MkString StartTime;
        it->second->GetStartTime().GetString(StartTime);
        JsonBody[Index]["startTime"] = StartTime;
        JsonBody[Index]["averageSpeedByte"] = CMkMediaUtil::GetAverageSpeed(it->second->GetRecvBytes(), it->second->GetStartTime());
        if (MkCodecUnknown != it->second->GetVideoParam().GetCodec()) {
            Json::Value JsonVideo;
            JsonVideo["frameRate"] = it->second->GetVideoParam().GetFrameRate();
            JsonVideo["width"] = it->second->GetVideoParam().GetWidth();
            JsonVideo["bitRate"] = it->second->GetVideoParam().GetBitRate();
            JsonVideo["height"] = it->second->GetVideoParam().GetHeight();
            JsonVideo["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetVideoParam().GetCodec());
            JsonBody[Index]["video"] = JsonVideo;
        }
        if (MkCodecUnknown != it->second->GetAudioParam().GetCodec()) {
            Json::Value JsonAudio;
            JsonAudio["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetAudioParam().GetCodec());
            JsonAudio["sampleRate"] = CMkMediaUtil::AudioSampleRateIndexToRate(it->second->GetAudioParam().GetSampleRate());
            JsonAudio["sampleSize"] = CMkMediaUtil::AudioSampleSizeIndexToSize(it->second->GetAudioParam().GetSampleSize());
            JsonAudio["bStereo"] = TRUE == it->second->GetAudioParam().GetStereo();
            JsonAudio["channel"] = it->second->GetAudioParam().GetChannel();
            JsonAudio["bitRate"] = it->second->GetAudioParam().GetBitRate();
            JsonBody[Index]["audio"] = JsonAudio;
        }
        it->second->StopMediaSource();
    }
    m_MapMediaSource.clear();
    m_MediaSourceLock.WUnlock();
    return WsNotifyMsg("MediaSourceClosed", JsonBody);
}

Uint32 CMkMediaServer::AddParser(const MkString& Path, const MkHttpMethod& Method, const MediaServerParser& Parser)
{
    Uint32 ErrorCode = NoneError;
    switch (Method) {
    case MkHttpMethodGet:
        m_MapGetParser[Path] = Parser;
        break;
    case MkHttpMethodPost:
        m_MapPostParser[Path] = Parser;
        break;
    case MkHttpMethodPut:
        m_MapPutParser[Path] = Parser;
        break;
    case MkHttpMethodDelete:
        m_MapDeleteParser[Path] = Parser;
        break;
    default:
        ErrorCode = UnSupported;
        break;
    }
    return ErrorCode;
}

MkString CMkMediaServer::GetRequestStringValue(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkString& strKey)
{
    MkString strValue = JsonReq[strKey].asString();
    if (strValue.empty()) {
        MkMapString::const_iterator it = MapQuery.find(strKey);
        if (it != MapQuery.end()) {
            strValue = it->second;
        }
    }
    return strValue;
}

Uint32 CMkMediaServer::GetRequestUint32Value(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkString& strKey)
{
    Uint32 uValue = 0;
    MkMapString::const_iterator it = MapQuery.find(strKey);
    if (it != MapQuery.end()) {
        uValue = CMkUtil::SafeAtoi(it->second.c_str());
    } else {
        uValue = JsonReq[strKey].asUInt();
    }
    return uValue;
}

float CMkMediaServer::GetRequestFloatValue(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkString& strKey)
{
    float fValue = 0.0f;
    MkMapString::const_iterator it = MapQuery.find(strKey);
    if (it != MapQuery.end()) {
        fValue = static_cast<float>(CMkUtil::SafeAtof(it->second.c_str()));
    } else {
        fValue = JsonReq[strKey].asFloat();
    }
    return fValue;
}

double CMkMediaServer::GetRequestDoubleValue(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkString& strKey)
{
    double dValue = 0.0;
    MkMapString::const_iterator it = MapQuery.find(strKey);
    if (it != MapQuery.end()) {
        dValue = CMkUtil::SafeAtof(it->second.c_str());
    } else {
        dValue = JsonReq[strKey].asFloat();
    }
    return dValue;
}

BOOL CMkMediaServer::GetRequestBooleanValue(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkString& strKey)
{
    BOOL bValue = FALSE;
    MkMapString::const_iterator it = MapQuery.find(strKey);
    if (it != MapQuery.end()) {
        bValue = "true" == CMkUtil::SafeMkString(it->second.c_str());
    } else {
        bValue = JsonReq[strKey].asBool();
    }
    return bValue;
}

Uint32 CMkMediaServer::GetVersionParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    MkString StrVersion;
    Uint32 ErrorCode = GetVersion(StrVersion);
    MkCheckErrorLog(ErrorCode, "get version failed\n");
    Json::Value ResBody;
    ResBody["version"] = StrVersion;
    ResCb(NoneError, ResBody);
    return NoneError;
}

Uint32 CMkMediaServer::GetConfigParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    Json::Value ResBody;
    MkVectorString VecAppNames;
    m_ConfigIni.GetProfileAppNames(VecAppNames);
    MkVectorString::iterator it = VecAppNames.begin();
    for (; it != VecAppNames.end(); it++) {
        MkVectorString VecKeyNames;
        m_ConfigIni.GetProfileKeyNames(*it, VecKeyNames);
        MkVectorString::iterator itKey = VecKeyNames.begin();
        for (; itKey != VecKeyNames.end(); itKey++) {
            ResBody[*it + "." + *itKey] = m_ConfigIni.GetProfileString(*it, *itKey, "");
        }
    }
    ResCb(NoneError, ResBody);
    return NoneError;
}

Uint32 CMkMediaServer::SetConfigParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    Json::Value::Members member = JsonReq.getMemberNames();
    Json::Value::Members::iterator iter = member.begin();

    Uint16 uHttpPort = CMkUtil::SafeAtoi(JsonReq[HttpPort].asCString());
    Uint16 uRtspPort = CMkUtil::SafeAtoi(JsonReq[RtspPort].asCString());
    Uint16 uRtmpPort = CMkUtil::SafeAtoi(JsonReq[RtmpPort].asCString());
    Uint16 uSipPort = CMkUtil::SafeAtoi(JsonReq[GbPort].asCString());
    Uint16 uRtpBeginPort = CMkUtil::SafeAtoi(JsonReq[RtpBeginPort].asCString());
    Uint16 uHttpPortBefore = m_ConfigIni.GetProfileUint32(HttpPort);
    Uint16 uRtspPortBefore = m_ConfigIni.GetProfileUint32(RtspPort);
    Uint16 uRtmpPortBefore = m_ConfigIni.GetProfileUint32(RtmpPort);
    Uint16 uRtpBeginPortBefore = m_ConfigIni.GetProfileUint32(RtpBeginPort);
    Uint16 uSipPortBefore = m_ConfigIni.GetProfileUint32(GbPort);
    for (; iter != member.end(); ++iter) {
        m_ConfigIni.SetProfileString(*iter, JsonReq[*iter].asString());
    }
    MkString ConfigPath;
    CMkFile::GetConfigPath(ConfigPath);
    MkString ConfigFile = ConfigPath + "/" + ExeName + ".ini";
    m_ConfigIni.WriteFile(ConfigFile);
    //reboot media server
    if (uHttpPort != uHttpPortBefore
        || uRtspPort != uRtspPortBefore
        || uRtmpPort != uRtmpPortBefore
        || uRtpBeginPortBefore != uRtpBeginPort
        || uSipPort != uSipPortBefore) {
        m_bReboot = TRUE;
        GetTimer()->LoopBreak();
        return NoneError;
    }

    m_pGbMediaServer->SetSipId(m_ConfigIni.GetProfileString(GeneralServerId));
    m_pGbMediaServer->SetTcpPublicHost(m_ConfigIni.GetProfileString(GbMappingHost));
    m_pGbMediaServer->SetUdpPublicHost(m_ConfigIni.GetProfileString(GbMappingHost));
    m_pGbMediaServer->SetTcpPublicPort(m_ConfigIni.GetProfileUint32(GbMappingPort));
    m_pGbMediaServer->SetTcpPublicPort(m_ConfigIni.GetProfileUint32(GbMappingPort));
    CMkSocket::SetTcpSendFailCumulativeMs(m_ConfigIni.GetProfileUint32(TcpUnSendMs, 3000));
    CMkSocket::SetTcpSendFailCumulativeMB(m_ConfigIni.GetProfileUint32(TcpUnSendMB, 20));
    CMkSocket::SetUdpSendFailCumulativeMB(m_ConfigIni.GetProfileUint32(UdpUnSendMB, 20));
    CMkSocket::SetUdpSendFailCumulativeMs(m_ConfigIni.GetProfileUint32(UdpUnSendMs, 3000));
    m_pHttpMediaServer->SetApiUriPrefix(m_ConfigIni.GetProfileString(HttpApiUriPrefix));
    m_pHttpMediaServer->SetKeepAliveSeconds(m_ConfigIni.GetProfileUint32(HttpKeepAliveSec, 60));
    m_pHttpMediaServer->SetMaxKeepAliveRequest(m_ConfigIni.GetProfileUint32(HttpMaxKeepAliveReq, 1000));
    m_pHttpMediaServer->SetUserPass(m_ConfigIni.GetProfileString(HttpUserName), m_ConfigIni.GetProfileString(HttpPassword));
    m_pHttpMediaServer->SetWebSocketPath(m_ConfigIni.GetProfileString(WsUrl));
    m_pHttpMediaServer->SetWebSocketAccessKey(m_ConfigIni.GetProfileString(WsAccessKey));
    m_pHttpMediaServer->SetMaxWsConnectorSize(m_ConfigIni.GetProfileUint32(WsMaxConnectorSize));
    return NoneError;
}

Uint32 CMkMediaServer::GetWsAccessKeyParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    Json::Value ResBody;
    ResBody["accessKey"] = m_pHttpMediaServer->GetWebSocketAccessKey();
    ResCb(NoneError, ResBody);
    return NoneError;
}

Uint32 CMkMediaServer::AddProxyStreamParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    MkString StreamPath = GetRequestStringValue(JsonReq, MapQuery, "streamPath");
    MkString ProxyUrl = GetRequestStringValue(JsonReq, MapQuery, "proxyUrl");
    if (StreamPath.empty()
        || ProxyUrl.empty()) {
        return InvalidParams;
    }
    MkString AppName = m_ConfigIni.GetProfileString(GeneralAppName);
    MkString MediaPath = "/" + AppName + "/" + StreamPath;
    CMkVideoParam VideoParam;
    CMkAudioParam AudioParam;
    Uint32 Duration;
    Uint32 FileSize;
    MkString    SourceSession;
    Uint32 ErrorCode = m_MediaCbs.GetSourceParamCb(MediaPath, VideoParam, AudioParam, Duration, FileSize, SourceSession);
    if (NoneError == ErrorCode) {  //is already has this path media source
        ResCb(InvalidUrl, Json::Value());
        return NoneError;
    }

    MkEventRemovedCallback RemoveCb = [this](CMkTcpClient* pClient) {
        CMkMediaSource *pSource = dynamic_cast<CMkMediaSource*>(pClient);
        if (pSource) {
            DeleteSource(pSource->GetSession());
        }
        pClient->GetEvent()->AddPipTask([pClient]() {
            delete pClient;
        });
    };

    MkSourceResultCallback ResultCb = [this, MediaPath, ResCb](CMkMediaSource* pMediaSource, Uint32 ErrorCode) {
        Json::Value ResBody;
        if (NoneError == ErrorCode) {
            pMediaSource->SetSession(CMkMd5Encode::Encode(MediaPath));
            OnMediaAdd(pMediaSource);
            ResBody["session"] = pMediaSource->GetSession();
        } else {
            CMkTcpClient *pClient = dynamic_cast<CMkTcpClient*>(pMediaSource);
            if (pClient) {
                pClient->GetEvent()->AddPipTask([pClient]() {
                    delete pClient;
                });
            }
        }
        ResCb(ErrorCode, ResBody);
        return NoneError;
    };

    if (CMkUtil::BeginByString(ProxyUrl, "rtmp")) {
        CMkRtmpPlayer* pSource = new CMkRtmpPlayer(m_pEventPool->GetFreeEvent(), m_pTimer, RemoveCb);
        pSource->SetAppName(AppName);
        pSource->SetStreamPath(StreamPath);
        pSource->StartMediaSource(ProxyUrl, ResultCb);
    } else if (CMkUtil::BeginByString(ProxyUrl, "rtsp")) {
        BOOL bTcpRtp = GetRequestBooleanValue(JsonReq, MapQuery, "bTcpRtp");
        CMkRtspPlayer* pSource = new CMkRtspPlayer(m_pEventPool->GetFreeEvent(), m_pTimer, m_pRtpPool->GetVecRtpUdpServer(), RemoveCb);
        pSource->SetAppName(AppName);
        pSource->SetStreamPath(StreamPath);
        pSource->Play(ProxyUrl, bTcpRtp, ResultCb);
    } else if (CMkUtil::BeginByString(ProxyUrl, "ws")) {
        CMkWsFlvPlayer *pSource = new CMkWsFlvPlayer(m_pEventPool->GetFreeEvent(), m_pTimer, RemoveCb);
        pSource->SetAppName(AppName);
        pSource->SetStreamPath(StreamPath);
        pSource->StartMediaSource(ProxyUrl, ResultCb);
    } else {
        MkErrorLog("input appName:%s streamPath:%s proxyUrl:%s\n", AppName.c_str(), StreamPath.c_str(), ProxyUrl.c_str());
        return UnSupportProtocol;
    }
    return NoneError;
}

Uint32 CMkMediaServer::GetMediaSourceParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    MkUnUse(JsonReq);
    MkUnUse(MapQuery);
    Json::Value ResBody;
    m_MediaSourceLock.RLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.begin();
    for (int i = 0; it != m_MapMediaSource.end(); it++, i++) {
        Json::Value Item;
        Item["appName"] = it->second->GetAppName();
        Item["streamPath"] = it->second->GetStreamPath();
        Item["sourceType"] = CMkMediaUtil::GetInstance().GetStringMediaSourceType(it->second->GetSourceType());
        Item["session"] = it->second->GetSession();
        Item["recvBytes"] = it->second->GetRecvBytes();
        MkString StartTime;
        it->second->GetStartTime().GetString(StartTime);
        Item["startTime"] = StartTime;
        Item["averageSpeedByte"] = CMkMediaUtil::GetAverageSpeed(it->second->GetRecvBytes(), it->second->GetStartTime());
        if (MkCodecUnknown != it->second->GetVideoParam().GetCodec()) {
            Json::Value JsonVideo;
            JsonVideo["frameRate"] = it->second->GetVideoParam().GetFrameRate();
            JsonVideo["width"] = it->second->GetVideoParam().GetWidth();
            JsonVideo["bitRate"] = it->second->GetVideoParam().GetBitRate();
            JsonVideo["height"] = it->second->GetVideoParam().GetHeight();
            JsonVideo["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetVideoParam().GetCodec());
            Item["video"] = JsonVideo;
        }
        if (MkCodecUnknown != it->second->GetAudioParam().GetCodec()) {
            Json::Value JsonAudio;
            JsonAudio["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetAudioParam().GetCodec());
            JsonAudio["sampleRate"] = CMkMediaUtil::AudioSampleRateIndexToRate(it->second->GetAudioParam().GetSampleRate());
            JsonAudio["sampleSize"] = CMkMediaUtil::AudioSampleSizeIndexToSize(it->second->GetAudioParam().GetSampleSize());
            JsonAudio["bStereo"] = it->second->GetAudioParam().GetStereo() ? true : false;
            JsonAudio["channel"] = it->second->GetAudioParam().GetChannel();
            JsonAudio["bitRate"] = it->second->GetAudioParam().GetBitRate();
            Item["audio"] = JsonAudio;
        }
        ResBody[i] = Item;
    }
    m_MediaSourceLock.RUnlock();
    ResCb(NoneError, ResBody);
    return NoneError;
}

Uint32 CMkMediaServer::GetMediaConumerParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    MkString Session = GetRequestStringValue(JsonReq, MapQuery, "session");
    MkString StreamPath = GetRequestStringValue(JsonReq, MapQuery, "streamPath");
    if (Session.empty()
        && StreamPath.empty()) {
        return InvalidParams;
    } else
        if (!StreamPath.empty()) {
            CMkVideoParam VideoParam;
            CMkAudioParam AudioParam;
            Uint32 Duration;
            Uint32 FileSize;
            OnMediaSourceGetParam("/" + m_ConfigIni.GetProfileString(GeneralAppName) + "/" + StreamPath, VideoParam, AudioParam, Duration, FileSize, Session);
        }
    if (Session.empty()) {
        return InvalidParams;
    }
    CMkTime NowTime = NowMkTime;
    Json::Value JsonBody;
    m_MediaSourceLock.RLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.find(Session);
    if (it != m_MapMediaSource.end()) {
        int Index = 0;
        MkMap<MkMediaConsumeType, CMkMediaConsume*>::const_iterator itConsume = it->second->GetMapConsumer().begin();
        for (; itConsume != it->second->GetMapConsumer().end(); itConsume++) {
            MkMap<MkString, CMkMediaConsumer*>::const_iterator itConsumer = itConsume->second->GetMapConsumer().begin();
            for (; itConsumer != itConsume->second->GetMapConsumer().end(); itConsumer++) {
                Json::Value Item;
                Item["ConsumerId"] = itConsumer->second->GetConsumerId();
                MkString StrTime;
                itConsumer->second->GetStartTime().GetString(StrTime);
                Item["StartTime"] = StrTime;
                Item["SendBytes"] = itConsumer->second->GetSendBytes();
                Item["AverageSpeedBytes"] = CMkMediaUtil::GetAverageSpeed(itConsumer->second->GetSendBytes(), itConsumer->second->GetStartTime(), NowTime);
                JsonBody[Index] = Item;
                Index++;
            }
        }
    }
    m_MediaSourceLock.RUnlock();
    ResCb(NoneError, JsonBody);
    return NoneError;
}

Uint32 CMkMediaServer::DeleteMediaSourceParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    MkString Session = GetRequestStringValue(JsonReq, MapQuery, "session");
    MkString StreamPath = GetRequestStringValue(JsonReq, MapQuery, "streamPath");
    if (Session.empty()
        && StreamPath.empty()) {
        return InvalidParams;
    } else
        if (!StreamPath.empty()) {
            CMkVideoParam VideoParam;
            CMkAudioParam AudioParam;
            Uint32 Duration;
            Uint32 FileSize;
            OnMediaSourceGetParam("/" + m_ConfigIni.GetProfileString(GeneralAppName) + "/" + StreamPath, VideoParam, AudioParam, Duration, FileSize, Session);
        }
    if (Session.empty()) {
        return InvalidParams;
    }

    if ("All" == Session) {
        ClearSource();
    } else {
        DeleteSource(Session);
    }
    ResCb(NoneError, Json::Value());
    return NoneError;
}

Uint32 CMkMediaServer::GetRtpPortRangeParse(const Json::Value& JsonReq, const MkMapString& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    Json::Value ResBody;
    if (m_pRtpPool) {
        ResBody["beginPort"] = m_pRtpPool->GetBeginPort();
        ResBody["endPort"] = m_pRtpPool->GetEndPort();
    }
    ResCb(NoneError, ResBody);
    return NoneError;
}

Uint32 CMkMediaServer::ShutDownParse(const Json::Value& JsonReq, const MkMap<MkString, MkString>& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    MkUnUse(JsonReq);
    MkUnUse(MapQuery);
    ResCb(NoneError, Json::Value());
    GetTimer()->LoopBreak();
    return NoneError;
}

Uint32 CMkMediaServer::RebootParse(const Json::Value& JsonReq, const MkMap<MkString, MkString>& MapQuery, const MkHttpResponseJsonCb& ResCb)
{
    MkUnUse(JsonReq);
    MkUnUse(MapQuery);
    ResCb(NoneError, Json::Value());
    m_bReboot = TRUE;
    GetTimer()->LoopBreak();
    return NoneError;
}

Uint32 CMkMediaServer::WsNotifyMsg(const MkString& strAction, const Json::Value& JsonBody, CMkHttpSession* pSession/* = nullptr*/)
{
    Json::Value NotifyJson;
    NotifyJson["action"] = strAction;
    NotifyJson["serverId"] = m_ConfigIni.GetProfileString(GeneralServerId);
    NotifyJson["data"] = JsonBody;
    if (nullptr == pSession) {
        m_pHttpMediaServer->BroadCastMsg(NotifyJson.toStyledString());
    } else {
        pSession->SendWebSocketText(NotifyJson.toStyledString());
    }
    return NoneError;
}

Uint32 CMkMediaServer::WsNotifyConfig(CMkHttpSession* pSession/* = nullptr*/)
{
    Json::Value JsonBody;
    MkVectorString VecAppNames;
    m_ConfigIni.GetProfileAppNames(VecAppNames);
    MkVectorString::iterator it = VecAppNames.begin();
    for (; it != VecAppNames.end(); it++) {
        MkVectorString VecKeyNames;
        m_ConfigIni.GetProfileKeyNames(*it, VecKeyNames);
        MkVectorString::iterator itKey = VecKeyNames.begin();
        for (; itKey != VecKeyNames.end(); itKey++) {
            JsonBody[*it + "." + *itKey] = m_ConfigIni.GetProfileString(*it, *itKey, "");
        }
    }
    return WsNotifyMsg("Config", JsonBody, pSession);
}

Uint32 CMkMediaServer::WsGbRegServerInfo(CMkHttpSession* pSession/* = nullptr*/)
{
    Json::Value JsonBody;
    JsonBody["bOnline"] = TRUE == m_pGbMediaServer->IsGbRegisterServerOnline();
    JsonBody["serverId"] = m_ConfigIni.GetProfileString(GbRegServerId);
    JsonBody["serverRealm"] = m_ConfigIni.GetProfileString(GbRegServerRealm);
    JsonBody["userName"] = m_ConfigIni.GetProfileString(GbRegUserName);
    JsonBody["password"] = m_ConfigIni.GetProfileString(GbRegPassword);
    JsonBody["host"] = m_ConfigIni.GetProfileString(GbRegHost);
    JsonBody["port"] = m_ConfigIni.GetProfileUint32(GbRegPort);
    JsonBody["bTcpSip"] = m_ConfigIni.GetProfileBoolean(GbTcpSip);
    return WsNotifyMsg("GbRegisterInfo", JsonBody, pSession);
}

Uint32 CMkMediaServer::WsNotifyMediaInfo(CMkHttpSession* pSession/* = nullptr*/)
{
    Json::Value JsonBody;
    m_MediaSourceLock.RLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.begin();
    for (int i = 0; it != m_MapMediaSource.end(); it++, i++) {
        Json::Value Item;
        Item["appName"] = it->second->GetAppName();
        Item["streamPath"] = it->second->GetStreamPath();
        Item["sourceType"] = CMkMediaUtil::GetInstance().GetStringMediaSourceType(it->second->GetSourceType());
        Item["session"] = it->second->GetSession();
        Item["recvBytes"] = it->second->GetRecvBytes();
        MkString StartTime;
        it->second->GetStartTime().GetString(StartTime);
        Item["startTime"] = StartTime;
        Item["averageSpeedByte"] = CMkMediaUtil::GetAverageSpeed(it->second->GetRecvBytes(), it->second->GetStartTime());
        if (MkCodecUnknown != it->second->GetVideoParam().GetCodec()) {
            Json::Value JsonVideo;
            JsonVideo["frameRate"] = it->second->GetVideoParam().GetFrameRate();
            JsonVideo["width"] = it->second->GetVideoParam().GetWidth();
            JsonVideo["bitRate"] = it->second->GetVideoParam().GetBitRate();
            JsonVideo["height"] = it->second->GetVideoParam().GetHeight();
            JsonVideo["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetVideoParam().GetCodec());
            Item["video"] = JsonVideo;
        }
        if (MkCodecUnknown != it->second->GetAudioParam().GetCodec()) {
            Json::Value JsonAudio;
            JsonAudio["mediaType"] = CMkMediaUtil::GetInstance().GetStringCodec(it->second->GetAudioParam().GetCodec());
            JsonAudio["sampleRate"] = CMkMediaUtil::AudioSampleRateIndexToRate(it->second->GetAudioParam().GetSampleRate());
            JsonAudio["sampleSize"] = CMkMediaUtil::AudioSampleSizeIndexToSize(it->second->GetAudioParam().GetSampleSize());
            JsonAudio["bStereo"] = it->second->GetAudioParam().GetStereo() ? true : false;
            JsonAudio["channel"] = it->second->GetAudioParam().GetChannel();
            JsonAudio["bitRate"] = it->second->GetAudioParam().GetBitRate();
            Item["audio"] = JsonAudio;
        }
        JsonBody[i] = Item;
    }
    m_MediaSourceLock.RUnlock();
    return WsNotifyMsg("MediaInfo", JsonBody);
}

CMkMediaServer::~CMkMediaServer()
{
    Stop();
    m_MediaSourceLock.WLock();
    MkMap<MkString, CMkMediaSource*>::iterator it = m_MapMediaSource.begin();
    for (; it != m_MapMediaSource.end(); it++) {
        MkDelete(it->second);
    }
    m_MapMediaSource.clear();
    m_MediaSourceLock.RUnlock();
    MkDelete(m_pHttpMediaServer);
    MkDelete(m_pRtmpServer);
    MkDelete(m_pGbMediaServer);
    MkDelete(m_pRtspServer);
    MkDelete(m_pRtpPool);
}
